import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.UserRegisterBean
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.FlinkSinkUtil
import io.circe._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import io.circe.generic.auto._
import io.circe.syntax._

import java.time.Duration

object UserRegister {


  def main(args: Array[String]): Unit = {
    implicit val conf = RunnerConfig("dws_user_user_register_window", TOPIC_DWD_USER_REGISTER, 10025, ckp = true, parallelism = 1)
    run { (_, ds) =>
      val registerStream = ds.flatMap { s =>
        for {
          json <- parser.parse(s).toOption
          jsonObj <- json.asObject
          id <- jsonObj("id")
          id <- id.asNumber
          _ <- id.toLong
          ts <- jsonObj("ts")
          ts <- ts.asNumber
          ts <- ts.toLong
        } yield ts * 1000 -> UserRegisterBean(1, ts * 1000)
      }
      val registerStreamWM = registerStream.assignTimestampsAndWatermarks(
        WatermarkStrategy.forBoundedOutOfOrderness(
          Duration.ofSeconds(3)).withTimestampAssigner(
            new SerializableTimestampAssigner[(Long, UserRegisterBean)] {
              override def extractTimestamp(element: (Long, UserRegisterBean), recordTimestamp: Long): Long = element._1
            }
          ))

      //      registerStreamWM.print()

      val aggStream = registerStreamWM
        .map(_._2)
        .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
        .reduce(_ + _, new AllWindowFunction[UserRegisterBean, UserRegisterBean, TimeWindow] {
          override def apply(window: TimeWindow, input: Iterable[UserRegisterBean], out: Collector[UserRegisterBean]): Unit = {
            import com.cw.realtime.common.util.TimeWindowUtil._
            val (stt, edt, cur) = window.windowRange
            input.foreach { e => out.collect(e.transWindowRange(stt, edt, cur)) }
          }
        })

//      aggStream.print()

      aggStream
        .map(_.asJson.noSpaces)
        .sinkTo(FlinkSinkUtil.getDorisSink(DWS_USER_USER_REGISTER_WINDOW))



    }


  }


}
